package com.cetc.sdp.kmga.cs.util;

import com.rabbitmq.client.*;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.TimeoutException;

/**
 * @description:
 * @author： DengQiang
 * @date: 2017/7/27 9:29
 */
public class RabbitMQProducer {
    //exchange type
    public enum XT {
        DEFAULT, DIRECT, TOPIC, HEADERS, FANOUT
    }

//    private String host;
//    private int port;
    private String userName;
    private String password;

    private String exchgName;
    private String queueName;

    private List<Address> addresses = new ArrayList<>();

    private Connection connection = null; //声明一个连接
    private Channel channel = null;
    private volatile XT xt = XT.DEFAULT;
    private String routingKey = "";

    private RabbitMQProducer(Map<String, Integer> ipPorts, String userName, String password, String exchgName, String queueName, XT xt) {
//        this.host = host;
//        this.port = port;
        this.userName = userName;
        this.password = password;
        this.exchgName = exchgName;
        this.queueName = queueName;
        this.xt = xt;
        ipPorts.forEach((ip, port) -> addresses.add(new Address(ip, port)));
        init();
    }

    public void setRoutingKey(String routingKey) {
        this.routingKey = routingKey;
    }

    private volatile static RabbitMQProducer instance;

    public static RabbitMQProducer getInstance(Map<String, Integer> ipPorts, String userName, String password, String exchgName, String queueName, XT xt) {

        if (instance == null) {
            synchronized (RabbitMQProducer.class) {
                if (instance == null) {
                    instance = new RabbitMQProducer(ipPorts, userName, password, exchgName, queueName, xt);
                }
            }
        }
        return instance;
    }

    private void init() {
        ConnectionFactory factory = new ConnectionFactory();
//        factory.setHost(host); //使用默认端口连接本地rabbitmq服务器
//        factory.setPort(port);
        factory.setUsername(userName);
        factory.setPassword(password);
        factory.setAutomaticRecoveryEnabled(true);

        try {
            connection = factory.newConnection(addresses);
            channel = connection.createChannel(); //声明消息通道
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
            return;
        }

//        connection.addShutdownListener(new ShutdownListener() {
//            @Override
//            public void shutdownCompleted(ShutdownSignalException cause) {
//                System.out.println("rabbitmq connection shutdown");
//                instance = null;
//                close();
//            }
//        });

        try {
            switch (xt) {
                case DEFAULT:
                    //默认，向指定的队列发送消息，消息只会被一个consumer处理,多个消费者消息会轮训处理,消息发送时如果没有consumer，消息不会丢失
                    //为消息通道绑定一个队列
                    //队列的相关参数需要与第一次定义该队列时相同，否则会出错
                    //参数1：队列名称
                    //参数2：为true时server重启队列不会消失
                    //参数3：队列是否是独占的，如果为true只能被一个connection使用，其他连接建立时会抛出异常
                    //参数4：队列不再使用时是否自动删除（没有连接，并且没有未处理的消息)
                    //参数5：建立队列时的其他参数
                    channel.queueDeclare(queueName, true, false, false, null);
                    break;
                case FANOUT:
                    //广播给所有队列  接收方也必须通过fanout交换机获取消息,所有连接到该交换机的consumer均可获取消息
                    //如果producer在发布消息时没有consumer在监听，消息将被丢弃

                    //定义一个交换机
                    //参数1：交换机名称
                    //参数2：交换机类型
                    //参数3：交换机持久性，如果为true则服务器重启时不会丢失
                    //参数4：交换机在不被使用时是否删除
                    //参数5：交换机的其他属性
                    channel.exchangeDeclare(exchgName, "fanout", true, false, null);
                    break;
                case DIRECT:
                    //向所有绑定了相应routing key的队列发送消息
                    //如果producer在发布消息时没有consumer在监听，消息将被丢弃
                    //如果有多个consumer监听了相同的routing key  则他们都会受到消息

                    channel.exchangeDeclare(exchgName, "direct", true, false, null);
                    channel.queueDeclare(queueName, true, false, false, null);
                    channel.queueBind(queueName, exchgName, routingKey);
                    break;
                case TOPIC:
                    //与direct模式有类似之处，都使用routing key作为路由
                    //不同之处在于direct模式只能指定固定的字符串，而topic可以指定一个字符串模式

                    channel.exchangeDeclare(exchgName, "topic", true, true, null);
                    channel.queueDeclare(queueName, false, false, false, null);
                    channel.queueBind(queueName, exchgName, routingKey);
                    break;
                case HEADERS:
                    //与topic和direct有一定相似之处，但不是通过routing key来路由消息
                    //通过headers中词的匹配来进行路由

                    channel.exchangeDeclare(exchgName, "headers", true, true, null);
                    break;
                default:
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void sendMsg(String message) {
        if (channel == null) {
            System.err.println("Cannot connect to rabbitmq");
            System.err.println("Send msg to rabbitmq failed: " + message);
            return;
        }
        try {
            switch (xt) {
                case DEFAULT:
                    //向server发布一条消息
                    //参数1：exchange名字，若为空则使用默认的exchange
                    //参数2：routing key
                    //参数3：其他的属性
                    //参数4：消息体
                    //RabbitMQ默认有一个exchange，叫default exchange，它用一个空字符串表示，它是direct exchange类型，
                    //任何发往这个exchange的消息都会被路由到routing key的名字对应的队列上，如果没有对应的队列，则消息会被丢弃
                    channel.basicPublish("", queueName, MessageProperties.TEXT_PLAIN, message.getBytes(Charset.forName("UTF-8"))); //设置消息为持久化，服务器重启不会丢失
                    break;
                case FANOUT:
                    channel.basicPublish(exchgName, routingKey, null, message.getBytes(Charset.forName("UTF-8")));
                    break;
                case DIRECT:
                    String[] temp = StringUtils.split(message, " ");
                    channel.basicPublish(exchgName, temp[0], MessageProperties.PERSISTENT_TEXT_PLAIN, temp[1].getBytes(Charset.forName("UTF-8")));
                    break;
                case TOPIC:
                    temp = StringUtils.split(message, " ");
                    channel.basicPublish(exchgName, temp[0], null, temp[1].getBytes(Charset.forName("UTF-8")));
                    break;
                case HEADERS:
                    //input like : headers message
                    temp = StringUtils.split(message, " ");
                    Map<String, Object> headers = new HashMap<String, Object>();
                    headers.put("name", temp[0]); //定义headers
                    headers.put("sex", temp[1]);
                    AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder().headers(headers);
                    channel.basicPublish(exchgName, routingKey, builder.build(), temp[2].getBytes(Charset.forName("UTF-8"))); //根据headers路由到相应的consumer
                    break;
                default:
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void close() {
        try {
            if (connection != null) {
                connection.close();
            }
            if (channel != null) {
                channel.close();
            }
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }

    }
}